package com.ratelimiter;

import com.Tread.SingleThreadPool;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicReferenceArray;
import java.util.concurrent.atomic.LongAdder;
public class SlowWindowTest {
    public static void main(String[] args) {
        ThreadPoolExecutor executor = SingleThreadPool.getInstance();
        //1s限流10次
        RateLimit rateLimit = new RateLimit(new ArrayDeque<>(), 10, 100L);
        for (int i = 0; i < 100; i++) {
            int finalI = i;
            executor.execute(() -> {
                if (rateLimit.tryAcquire()) {
                    System.out.println(Thread.currentThread() + "success.." + finalI);

                }else  {
                    System.out.println(Thread.currentThread() + "limit!!!!!");
                }
            });

        }

    }
}


/**
 * 可以查看历史某一段时间，请求数；qps限制为10
 * //保存最近1分钟的数据
 */
class SentinelRateLimit {

    //默认500ms
    long windowSize;

    AtomicReferenceArray<Bucket> buckets;

    int limitReqNum;

    //数组的大小取决于要保存多少时间窗口
    public SentinelRateLimit(long windowSize, int rateLimit) {
        this.windowSize = windowSize;
        this.buckets = new AtomicReferenceArray<>(1024);
    }

    public boolean tryAcquire() {
        long timeMillis = System.currentTimeMillis();
        int windowIndex = (int) ((timeMillis / 500) % buckets.length());
        Bucket bucket = new Bucket(timeMillis);

        if (buckets.compareAndSet(windowIndex, null, bucket)) {
               bucket.getCurReqNum().add(1L);
        } else {

            if ((buckets.get(windowIndex).getCurReqNum().sum() + buckets.get(windowIndex - 1).getCurReqNum().sum()) >
                    this.limitReqNum ) {
                System.out.println("限流触发>>>>");
                return false;
            }
            else {
                buckets.get(windowIndex).getCurReqNum().add(1L);
                return true;
            }
        }

        return true;
    }
}

class Bucket {

    LongAdder curReqNum;

    LongAdder errNum;

    long startTime;

    public Bucket(long startTime) {
        this.startTime = startTime;
        this.curReqNum = new LongAdder();
    }

    public LongAdder getCurReqNum() {
        return curReqNum;
    }

    public void setCurReqNum(LongAdder curReqNum) {
        this.curReqNum = curReqNum;
    }

    public LongAdder getErrNum() {
        return errNum;
    }

    public void setErrNum(LongAdder errNum) {
        this.errNum = errNum;
    }

    public long getStartTime() {
        return startTime;
    }

    public void setStartTime(long startTime) {
        this.startTime = startTime;
    }
}

/**
 * 标准滑动窗口
 */
class RateLimit {
    //请求的时间戳
    Deque<Long> queue;
    int ct;

    Long windowSize ;

    public RateLimit(Deque<Long> queue, int ct, Long windowSize) {
        this.queue = queue;
        this.ct = ct;
        this.windowSize = windowSize;
    }

    public synchronized boolean tryAcquire() {
        long currentTimeMillis = System.currentTimeMillis();
        long preTimeMillis =  currentTimeMillis - windowSize;

        //先移除一个时间窗口之前的所有元素
        while (queue.size() != 0) {
            if (preTimeMillis > queue.peekFirst()) {
                queue.pollFirst();
            } else {
                break;
            }
        }
        //统计当前元素
        if (queue.size() >= this.ct) {
                return false;
        }
        queue.offerLast(currentTimeMillis);
        return true;
    }
}
